Kafka Streams count throwing Exception